package com.pig4cloud.pigx.contract.thread;

import com.mongodb.BasicDBObject;
import com.mongodb.client.model.IndexModel;
import com.pig4cloud.pigx.common.customize.dto.TimeRangeDTO;
import com.pig4cloud.pigx.common.customize.mongo.utils.MongoUtils;
import com.pig4cloud.pigx.common.customize.redis.utils.RedisUtils;
import com.pig4cloud.pigx.common.customize.thread.SoftEndThread;
import com.pig4cloud.pigx.common.customize.utils.CommonUtil;
import com.pig4cloud.pigx.common.customize.utils.id.IdWorker;
import com.pig4cloud.pigx.contract.bo.CalculateKLineThreadParamBO;
import com.pig4cloud.pigx.contract.constant.ContractKLineConstants;
import com.pig4cloud.pigx.contract.constant.ContractRedisKeyConstants;
import com.pig4cloud.pigx.contract.entity.ContractKLine;
import com.pig4cloud.pigx.contract.enums.ContractOrderEnum;
import com.pig4cloud.pigx.contract.mapper.ContractKLineMapper;
import com.pig4cloud.pigx.contract.mapper.ContractTradeMapper;
import com.pig4cloud.pigx.contract.mongo.entity.ContractKLineMongo;
import com.pig4cloud.pigx.contract.mongo.entity.ContractKLineTradeCacheMongo;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.thread.repair.RepairKLine;
import com.pig4cloud.pigx.contract.thread.rule.SaveTradeRule;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.CollectionOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.support.CronSequenceGenerator;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * 计算k线数据线程
 *
 * @date 2022/8/15 19:48
 */
@Data
@Slf4j
@AllArgsConstructor
public class CalculateKLineThread extends SoftEndThread {
    private final ContractTradeMapper contractTradeMapper;

    private final CalculateKLineThreadParamBO param;
    private final CommonUtil commonUtil;
    private final RedisUtils redisUtils;
    private final ContractUtils utils;
    private final RedisTemplate redis;
    private final MongoTemplate mongo;
    private final AmqpTemplate rabbit;
    private final IdWorker idWorker;
    private final ContractKLineMapper lineMapper;

    private final RepairKLine repairKLine;

    /**
     * 一个k线周期一个线程
     *
     * @date 2022/8/15 19:57
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void run() {
        while (SoftEndThread.runnable) {
            // 根据timeRangeType获得相关配置
            ContractKLineConstants.TimeRangeTypeConfig config = ContractKLineConstants.findByTimeRangeType(param.getTimeRangeType());
            if (Objects.isNull(config)) {
                String errorMsg = "数据异常，未能根据timeRangeType：%s 查询到对应的TimeRangeTypeConfig";
//				log.error(errorMsg);
                throw new RuntimeException(errorMsg);
            }
            String lockKey = ContractRedisKeyConstants.CALCULATE_K_LINE_LOCK + "_" + config.getType();
            // 争夺redis锁
            RLock lock = redisUtils.getLock(lockKey, null);
            if (Objects.isNull(lock)) {
                return;
            }

            try {
                // 根据timeRangeType获得开始日期和结束日期
                LocalDateTime now = LocalDateTime.now();
                TimeRangeDTO timeRange = getTimeRange(config, now);
                // 组装查询语句的开始时间-结束时间
                TimeRangeDTO selectTimeRange = new TimeRangeDTO();
                // 以当前方法执行时间为结束时间
                selectTimeRange.setEndTime(now);
                // 查询时间根据redis记录的当前K线最后执行时间来获取，如果没有值，则以当前所属时间段开始时间作为查询开始时间
                LocalDateTime lastDate = timeRange.getStartTime();
                Boolean hasKey = redis.hasKey(config.getType());
                if (hasKey) {
                    lastDate = (LocalDateTime) redis.opsForValue().get(config.getType());
                }
                // 设置查询结束时间
                selectTimeRange.setStartTime(lastDate);
                // redis 覆盖 最后执行时间
                redis.opsForValue().set(config.getType(), now);
                // 根据时间范围，查询k线数据
                // mysql查询 淘汰
//				List<ContractKLineVO> list = contractTradeMapper.selectForCalculateKLineThread(selectTimeRange);
                // mongoDB分组查询 淘汰
//				String collectionName = "k_line_trade_cache";
//				Aggregation aggregation = Aggregation.newAggregation(
//						// 查询条件
//						Aggregation.match(
//								Criteria.where("createTime")
//										.gte(selectTimeRange.getStartTime().plusSeconds(-1))
//										.lt(selectTimeRange.getEndTime().plusSeconds(-1))
//						),
//						Aggregation
//								// 根据币对id分组
//								.group("symbolId")
//								// 聚合函数字段
//								.first("price").as("price")
//								.max("price").as("maxPrice")
//								.first("price").as("price")
//								.min("price").as("minPrice")
//								.first("num").as("num")
//								.sum("num").as("sumNum")
//								// 查询基础字段
//								.first("symbolId").as("symbolId")
//								.first("symbolName").as("symbolName")
//								.first("createTime").as("createTime")
//				);
//				AggregationResults<ContractKLineVO> kLineVOS = mongo.aggregate(aggregation, collectionName, ContractKLineVO.class);
//				List<ContractKLineVO> list = kLineVOS.getMappedResults();
                // 目前选用自造数据，后续根据情况可能选用上面两种情景
                List<ContractSymbolMongo> mongoList = mongo.find(Query.query(Criteria.where("status").is("1"))
                        , ContractSymbolMongo.class);
                // 循环数据 每条代表一个币种
                for (ContractSymbolMongo vo : mongoList) {
                    if (ContractKLineConstants.SAVE_TRADE_RANGE_TYPE_LIST.contains(config.getType())) {
                        // 虚拟最新成交记录 配合K线展示
                        this.saveTrade(vo);
                    }
                    // 数据缓存至mongodb;
                    // mongodb创建新时间节点数据时,将上一完结时间节点数据更新至mysql、
                    this.saveMongoDb(vo, config, timeRange);
                }

                Thread.sleep(500L);

                // 释放redis锁
                redisUtils.releaseLock(lock);
            } catch (Exception ignore) {
            } finally {
                // 释放redis锁
                redisUtils.releaseLock(lock);
            }
        }

        end();
    }

    /**
     * 根据时间范围类型和当前时间获得当前时间所在的时间段
     *
     * @param config 日期范围类型设置
     * @param now    当前时间
     * @return {@link TimeRangeDTO}
     * @date 2022/8/15 17:10
     */
    private TimeRangeDTO getTimeRange(ContractKLineConstants.TimeRangeTypeConfig config, LocalDateTime now) {
        // System.out.println("currentTime: " + now);

        CronSequenceGenerator cronSequenceGenerator = new CronSequenceGenerator(config.getCron());

        // 下个时间节点作为结束时间
        Date endTime = cronSequenceGenerator.next(this.localDateTimeToDate(now));
        // System.out.println("nextTimePoint: " + endTime);

        // 将下个时间节点作为结束时间，逆推开始时间
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(endTime);
        calendar.add(config.getCalendarMinusField(), config.getCalendarMinusNum());

        TimeRangeDTO timeRange = new TimeRangeDTO();
        timeRange.setStartTime(this.dateToLocalDateTime(calendar.getTime()));
        timeRange.setEndTime(this.dateToLocalDateTime(endTime));
        return timeRange;
    }

    /**
     * 异步生成交易记录
     *
     * @param symbolMongo 币对详情
     */
    @Async
    public void saveTrade(ContractSymbolMongo symbolMongo) {
        // 以币对名称设置集合名，没有集合则创建
        String collectionName = symbolMongo.getName() + "_newest_trade";
        boolean exists = mongo.collectionExists(collectionName);
        if (!exists) {
            // 集合封顶字节大小
            Long size = 1073741824L;
            // 集合封顶数据条数
            Long maxDocuments = 20L;
            // 新建集合
            mongo.createCollection(collectionName, new CollectionOptions(size, maxDocuments, true));
        }
        Query query = new Query();
        query.with(Sort.by(Sort.Order.desc("createTime")));
        query.skip(0);
        query.limit(1);
        ContractKLineTradeCacheMongo vo =
                mongo.findOne(query, ContractKLineTradeCacheMongo.class, collectionName);
        if (vo == null) {
            vo = new ContractKLineTradeCacheMongo();
            vo.setSymbolId(symbolMongo.getId());
            vo.setSymbolName(symbolMongo.getName());
            vo.setCreateTime(LocalDateTime.now());
            vo.setSide(ContractOrderEnum.Side.MORE.getValue());
        } else {
            vo.setCreateTime(LocalDateTime.now());
            // 根据行情涨跌状况,判断挂空单还是多单
            if (symbolMongo.getLastPrice().compareTo(vo.getPrice()) > 0) {
                vo.setSide(ContractOrderEnum.Side.MORE.getValue());
            }
            if (symbolMongo.getLastPrice().compareTo(vo.getPrice()) == 0) {
                double random = Math.random() * 10;
                if (random <= 5) {
                    vo.setSide(ContractOrderEnum.Side.EMPTY.getValue());
                } else {
                    vo.setSide(ContractOrderEnum.Side.MORE.getValue());
                }
            }
            if (symbolMongo.getLastPrice().compareTo(vo.getPrice()) < 0) {
                vo.setSide(ContractOrderEnum.Side.EMPTY.getValue());
            }
        }
        // 生成随机成交数量
        BigDecimal num = SaveTradeRule.getNum(
                symbolMongo.getLastPrice(),
                symbolMongo.getMinOrderNum(),
                symbolMongo.getMaxMarketOrderNum()
        );
        vo.setNum(num);
        // 价格选用最新价
        vo.setPrice(symbolMongo.getLastPrice());
        vo.setTradeValue(vo.getNum().multiply(vo.getPrice()));
        // 上限20条，直接插入覆盖
        mongo.insert(vo, collectionName);
        // 24小时交易记录缓存
        save24hTrade(vo);
    }

    /**
     * 新增24小时成交记录
     *
     * @param vo
     */
    public void save24hTrade(ContractKLineTradeCacheMongo vo) {
        String collectionName = vo.getSymbolName() + "_newest_trade_24h";
        boolean exists = mongo.collectionExists(collectionName);
        if (!exists) {
            mongo.createCollection(collectionName);
        }
        mongo.insert(vo, collectionName);
    }


    /**
     * 根据定时查询k线数据 新增/修改 mongo相关数据
     *
     * @param vo        k线数据
     * @param config    时间段配置
     * @param timeRange 该时间段的起止时间
     */
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.SUPPORTS)
    public void saveMongoDb(ContractSymbolMongo vo,
                            ContractKLineConstants.TimeRangeTypeConfig config,
                            TimeRangeDTO timeRange
    ) {
        // 集合名称 规则：以 (币对id_k线类型) 命名，确定唯一性
        String collectionName = vo.getName() + "_" + config.getType();
        // 查询该币种在该K线上是否有collection集合 如果没有则进行创建
        boolean exists = mongo.collectionExists(collectionName);
        if (!exists) {
            // 集合封顶字节大小,该变量可放入配置文件中
            Long size = 1073741824L;
            // 集合封顶数据条数，该变量可放入配置文件中
            Long maxDocuments = 500L;
            //索引
            List<IndexModel> indexModels = new ArrayList<>();
            BasicDBObject index1 = new BasicDBObject();
            index1.put("startTime", 1);
            index1.put("endTime", 1);
            indexModels.add(new IndexModel(index1));
            BasicDBObject index2 = new BasicDBObject();
            index2.put("startTime", 1);
            indexModels.add(new IndexModel(index2));
            BasicDBObject index3 = new BasicDBObject();
            index3.put("endTime", 1);
            indexModels.add(new IndexModel(index3));
            // 新建集合
            mongo.createCollection(collectionName, new CollectionOptions(size, maxDocuments, true)).createIndexes(indexModels);
        }
        // 查询该时间段的数据有无缓存记录，有则修改，无则删除
        Criteria criteria = Criteria
                .where("startTime").is(timeRange.getStartTime())
                .and("endTime").is(timeRange.getEndTime());
        List<ContractKLineMongo> lineMongoList =
                mongo.find(Query.query(criteria), ContractKLineMongo.class, collectionName);
        // 组装币对id 和 k线时间段 用于查询开盘价和收盘价
        TimeRangeDTO priceTime = new TimeRangeDTO();
        BeanUtils.copyProperties(timeRange, priceTime);
        priceTime.setSymbolId(vo.getId());
        // 查询币对最新价
        BigDecimal lastPrice = vo.getLastPrice();
        if (lineMongoList.size() > 0) {
            ContractKLineMongo lineMongo = lineMongoList.get(0);
            // 判断先后两次数据的最高价大小,留下最高价最高的
            int compare = lineMongo.getHighPrice().compareTo(lastPrice);
            if (compare < 0) {
                lineMongo.setHighPrice(lastPrice);
            }
            // 判断先后两次数据的最低价大小,留下最高价最低的
            compare = lineMongo.getLowPrice().compareTo(lastPrice);
            if (compare > 0) {
                lineMongo.setLowPrice(lastPrice);
            }
            // 将两次数据中的成交量合并
            BigDecimal num = SaveTradeRule.getNum(
                    vo.getLastPrice(),
                    vo.getMinOrderNum(),
                    vo.getMaxMarketOrderNum()
            );
            lineMongo.setDealNum(lineMongo.getDealNum().add(num).setScale(vo.getTokenCurAccuracy(), RoundingMode.HALF_UP));
            // 成交额
            if (lineMongo.getDealPrice() != null) {
                BigDecimal multiply = num.multiply(lastPrice);
                lineMongo.setDealPrice(lineMongo.getDealPrice().add(multiply).setScale(vo.getMainCurAccuracy(), RoundingMode.HALF_UP));
            }
            // 编辑时只需要改收盘价 不需要变动开盘价
            lineMongo.setClosePrice(lastPrice);
            // 修改条件
            Query query = Query.query(Criteria.where("_id").is(lineMongo.getId()));
            // 修改内容
            Update update = MongoUtils.getUpdate(lineMongo);
            mongo.updateMulti(query, update, collectionName);
            // log.error(collectionName + " mongo更新-------------");
        } else {
            //***** 产生新节点的时候 , 查询是否能正确衔接上一时间节点 如果衔接不上 或者 上一时间节点数据异常 , 此时进行K线修复工作******
            ContractKLineMongo mongoOne = mongo.findOne(new Query() {{
                skip(0);
                limit(1);
                with(Sort.by(Sort.Order.desc("startTime")));
            }}, ContractKLineMongo.class, collectionName);
            if (mongoOne != null) {
                try {
                    long startTimeStamp = RepairKLine.toTimeStamp(timeRange.getStartTime());
                    long endTimeStamp = RepairKLine.toTimeStamp(mongoOne.getEndTime());
                    // 时间无法衔接的情况,进行数据填充操作
                    if (startTimeStamp != endTimeStamp) {
                        ContractKLineMongo kLineMongo = repairKLine.repairForInsert(
                                mongoOne.getSymbolId(),
                                mongoOne.getSymbolName(),
                                config.getBiAnLineType(),
                                endTimeStamp,
                                startTimeStamp,
                                collectionName,
                                config.getTableName()
                        );
                        mongoOne.setClosePrice(kLineMongo.getOpenPrice());
                        Query query = Query.query(Criteria.where("startTime").is(mongoOne.getStartTime()));
                        // 修改内容
                        Update update = MongoUtils.getUpdate(mongoOne);
                        mongo.updateMulti(query, update, collectionName);
                        return;
                    } else {
                        // 数据为一条横线的异常情况时,进行数据修复操作
                        if (mongoOne.getHighPrice().compareTo(mongoOne.getLowPrice()) == 0) {
                            mongoOne = repairKLine.repairForUpdate(
                                    mongoOne.getSymbolId(),
                                    mongoOne.getSymbolName(),
                                    config.getBiAnLineType(),
                                    RepairKLine.toTimeStamp(mongoOne.getStartTime()),
                                    collectionName
                            );
                        }
                    }
                    // mongo产生新时间节点数据时,将上一时间节点完整数据插入至mysql
                    String tableName = config.getTableName();
                    ContractKLine kLine = new ContractKLine();
                    BeanUtils.copyProperties(mongoOne, kLine);
                    // 表名
                    kLine.setTableName(tableName);
                    // 新增
                    lineMapper.insertContractKLine(kLine);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            ContractKLineMongo lineMongo = new ContractKLineMongo();
            // 主键
            lineMongo.setId(idWorker.nextId());
            // 最高价
            lineMongo.setHighPrice(lastPrice);
            // 最低价
            lineMongo.setLowPrice(lastPrice);
            // 成交量
            BigDecimal num = SaveTradeRule.getNum(
                    vo.getLastPrice(),
                    vo.getMinOrderNum(),
                    vo.getMaxMarketOrderNum()
            );
            lineMongo.setDealNum(num.setScale(vo.getTokenCurAccuracy(), RoundingMode.HALF_UP));
            // 成交额
            BigDecimal multiply = num.multiply(lastPrice);
            lineMongo.setDealPrice(multiply.setScale(vo.getMainCurAccuracy(), RoundingMode.HALF_UP));
            // 币对id
            lineMongo.setSymbolId(vo.getId());
            // 币对名
            lineMongo.setSymbolName(vo.getName());
            // 开始时间
            lineMongo.setStartTime(timeRange.getStartTime());
            // 结束时间
            lineMongo.setEndTime(timeRange.getEndTime());
            // 获取上一节点收盘价,定义为本次开盘价
            Query query = new Query();
            query.with(Sort.by(Sort.Order.desc("startTime")));
            query.skip(0);
            query.limit(1);
            ContractKLineMongo cacheMongo =
                    mongo.findOne(query, ContractKLineMongo.class, collectionName);
            if (cacheMongo != null) {
                lineMongo.setOpenPrice(cacheMongo.getClosePrice());
            } else {
                lineMongo.setOpenPrice(lastPrice);
            }
            // 收盘价
            lineMongo.setClosePrice(lineMongo.getOpenPrice());
            // 插入 (集合数据条数有上限，溢出数据自动删除)
            mongo.insert(lineMongo, collectionName);
            // log.error(collectionName + " mongo插入-------------");
        }
    }

    /**
     * 根据定时查询k线数据 新增/修改 mysql相关数据
     *
     * @param vo        k线数据
     * @param config    时间段配置
     * @param timeRange 该时间段的起止时间
     */
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.SUPPORTS)
    public void saveMysql(ContractSymbolMongo vo,
                          ContractKLineConstants.TimeRangeTypeConfig config,
                          TimeRangeDTO timeRange
    ) {
        // 获取当前时间段所使用的数据库表名
        String tableName = config.getTableName();
        ContractKLine line = new ContractKLine();
        line.setTableName(tableName);
        line.setStartTime(timeRange.getStartTime());
        line.setEndTime(timeRange.getEndTime());
        line.setSymbolId(vo.getId());
        // 查询当前时间段 当前币对 是否有数据 有则合并覆盖，无则新增
        List<ContractKLine> lines = lineMapper.selectByTime(line);
        // 组装币对id 和 k线时间段 用于查询开盘价和收盘价
        TimeRangeDTO priceTime = new TimeRangeDTO();
        BeanUtils.copyProperties(timeRange, priceTime);
        priceTime.setSymbolId(vo.getId());
        // 查询币对最新价
        BigDecimal lastPrice = vo.getLastPrice();
        if (lines.size() > 0) {
            ContractKLine kLine = lines.get(0);
            kLine.setTableName(tableName);
            // 判断先后两次数据的最高价大小,留下最高价最高的
            int compare = kLine.getHighPrice().compareTo(lastPrice);
            if (compare < 0) {
                kLine.setHighPrice(lastPrice);
            }
            // 判断先后两次数据的最低价大小,留下最高价最低的
            compare = vo.getMinPrice().compareTo(lastPrice);
            if (compare < 0) {
                kLine.setLowPrice(lastPrice);
            }
            // 合并成交量
            kLine.setDealNum(kLine.getDealNum().add(lastPrice));
            // 编辑时只需要查收盘价 不需要变动开盘价
            kLine.setClosePrice(lastPrice);
            // 修改
            lineMapper.updateContractKLine(kLine);
            // log.error(tableName + " mysql更新-------------");
        } else {
            ContractKLine kLine = new ContractKLine();
            kLine.setTableName(tableName);
            kLine.setId(idWorker.nextId());
            kLine.setStartTime(timeRange.getStartTime());
            kLine.setEndTime(timeRange.getEndTime());
            kLine.setHighPrice(lastPrice);
            kLine.setLowPrice(lastPrice);
            kLine.setDealNum(lastPrice);
            kLine.setSymbolId(vo.getId());
            kLine.setSymbolName(vo.getName());
            // 获取上一节点收盘价,定义为本次开盘价
            ContractKLine contractKLine = utils.selectCurrentKLine(kLine.getSymbolName(), config.getType());
            if (contractKLine != null && contractKLine.getClosePrice() != null) {
                kLine.setOpenPrice(contractKLine.getClosePrice());
            } else {
                kLine.setOpenPrice(lastPrice);
            }
            // 收盘价
            kLine.setClosePrice(kLine.getOpenPrice());
            // 新增
            lineMapper.insertContractKLine(kLine);
            // log.error(tableName + " mysql插入-------------");
        }
    }

    /**
     * 时间转换 Date -> LocalDateTime
     *
     * @param date
     * @return
     */
    private LocalDateTime dateToLocalDateTime(Date date) {
        return Instant.ofEpochMilli(date.getTime())
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
    }

    /**
     * 时间转换 LocalDateTime -> Date
     *
     * @param localDateTime
     * @return
     */
    private Date localDateTimeToDate(LocalDateTime localDateTime) {
        return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
    }
}
